package com.hsogoo.avatar.executor.disruptor;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import com.hsogoo.avatar.NamedThreadFactory;
import com.hsogoo.avatar.constants.AvatarConstant;
import com.hsogoo.avatar.executor.AvatarExecutor;
import com.lmax.disruptor.BlockingWaitStrategy;
import com.lmax.disruptor.EventFactory;
import com.lmax.disruptor.InsufficientCapacityException;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.WaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;

/**
 * Created by za-huangsaigang on 2018/10/29.
 */
public class AvatarDisruptor implements AvatarExecutor {

    private static int BUFFER_SIZE = 1024;
    private static int MAX_NUM_WORKERS = AvatarConstant.AVAILABLE_PROCESSORS << 2;

    private static final EventFactory<AvatarEvent<Runnable>> eventFactory = new EventFactory<AvatarEvent<Runnable>>() {

        @Override
        public AvatarEvent<Runnable> newInstance() {
            return new AvatarEvent<>();
        }
    };

    private final Disruptor<AvatarEvent<Runnable>> disruptor;
    private final ExecutorService reserveExecutor;

    public AvatarDisruptor(int numWorkers,
                           ThreadFactory threadFactory){
        this(numWorkers, threadFactory, BUFFER_SIZE, 0 , new BlockingWaitStrategy(), null);
    }

    public AvatarDisruptor(int numWorkers,
                           ThreadFactory threadFactory,
                           int bufSize,
                           int numReserveWorkers,
                           WaitStrategy waitStrategyType,
                           String dumpPrefixName) {

        if (numReserveWorkers > 0) {
            String name = "reserve.processor";
            reserveExecutor = new ThreadPoolExecutor(0, numReserveWorkers, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new NamedThreadFactory(name),  new ThreadPoolExecutor.AbortPolicy());
        } else {
            reserveExecutor = null;
        }
        if (threadFactory == null) {
            threadFactory = new NamedThreadFactory("disruptor.processor");
        }
        Disruptor<AvatarEvent<Runnable>> dr =
                new Disruptor<>(eventFactory, bufSize, threadFactory, ProducerType.MULTI, new BlockingWaitStrategy());
//        dr.setDefaultExceptionHandler(new LoggingExceptionHandler());
        numWorkers = Math.min(Math.abs(numWorkers), MAX_NUM_WORKERS);
        if (numWorkers == 1) {
            dr.handleEventsWith(new TaskHandler());
        } else {
            TaskHandler[] handlers = new TaskHandler[numWorkers];
            for (int i = 0; i < numWorkers; i++) {
                handlers[i] = new TaskHandler();
            }
            dr.handleEventsWithWorkerPool(handlers);
        }

        dr.start();
        disruptor = dr;
    }

    private boolean dispatch(Runnable task) {
        RingBuffer<AvatarEvent<Runnable>> ringBuffer = disruptor.getRingBuffer();
        try {
            long sequence = ringBuffer.tryNext();
            try {
                AvatarEvent<Runnable> event = ringBuffer.get(sequence);
                event.setTask(task);
            } finally {
                ringBuffer.publish(sequence);
            }
            return true;
        } catch (InsufficientCapacityException e) {
            // 这个异常是Disruptor当做全局goto使用的, 是单例的并且没有堆栈信息, 不必担心抛出异常的性能问题
            return false;
        }
    }

    @Override
    public void execute(Runnable task) {
        if (!dispatch(task)) {
            // 备选线程池
            if (reserveExecutor != null) {
                reserveExecutor.execute(task);
            } else {
                throw new RejectedExecutionException("Ring buffer is full");
            }
        }
    }

    @Override
    public void shutdown() {
        disruptor.shutdown();
        if (reserveExecutor != null) {
            reserveExecutor.shutdownNow();
        }
    }
}
